package com.dong.flink;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by dong peng on 2020/8/8.
 */
public class WordCountDemo {

    public static void count() {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        Tuple2<String, Integer> t1 = new Tuple2<>("a", 1);
        Tuple2<String, Integer> t2 = new Tuple2<>("b", 2);
        Tuple2<String, Integer> t3 = new Tuple2<>("a", 3);
        Tuple2<String, Integer> t4 = new Tuple2<>("b", 3);
        Tuple2<String, Integer> t5 = new Tuple2<>("b", 4);
        DataStream dataStream = environment.fromElements(t1,t2,t3,t4,t5);

        dataStream.keyBy(0).reduce(new ReduceFunction() {
            @Override
            public Object reduce(Object o, Object t1) throws Exception {
                System.out.println("对象o: "  + o + "  对象t: " + t1);
                Tuple2<String, Integer> o1 = (Tuple2) o;
                Tuple2<String, Integer> t2 = (Tuple2) t1;
                return new Tuple2(o1.f0, o1.f1 + t2.f1);
            }
        }).print();

        try {
            environment.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void nettyTest() {
        /**
         * bossGroup, 父类的事件循环组只是负责连接，获取到连接后交给 workergroup子的事件循环组，
         * 参数的获取，业务的处理等工作均是由workergroup这个子事件循环组来完成，一个事件循环组一样
         * 可以完成所有的工作，但是Netty推荐的方式是使用两个事件循环组。
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();  //创建父事件循环组
        EventLoopGroup workerGroup = new NioEventLoopGroup(); //创建子类的事件循环组

        try{
            //创建启动服务器的对象
            ServerBootstrap serverBootstrap = new ServerBootstrap();

            /**
             * group方法接收两个参数， 第一个为父时间循环组，第二个参数为子事件循环组
             */
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer() {
                @Override
                protected void initChannel(Channel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    /**
                     * Handler就相当于Servlet中的过滤器, 请求和响应都会走Handler
                     * HttpServerCodec: http的编解码器，用于Http请求和相应
                     */
                    pipeline.addLast("httpServerCodec", new HttpServerCodec());
                    pipeline.addLast("testHttpServerHandler", new SimpleChannelInboundHandler() {
                        @Override
                        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
                            System.out.println(channelHandlerContext);
                        }
                    });
                }
            }); //bossGroup的通道，只是负责连接

            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();  //绑定端口
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally{
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}
